/*
* Created by boil on 2023/10/26.
*/

#include "event_loop.h"
#include "event_loop_pool.h"
#include "net/utils/socket_helper.h"
#include "time/time_ticker.h"
#include "select_wrap.h"
#include "utils/rd_errno.h"

RD_NAMESPACE_BEGIN

  EventLoop &EventLoop::Instance() {
    return *(EventLoopPool::Instance().getFirstPoller());
  }

  void EventLoop::addEventPipe() {
    SocketHelper::SetNoBlocked(_pipe.ReadFD());
    SocketHelper::SetNoBlocked(_pipe.WriteFD());

    // 添加内部管道事件
    if (AddEvent(_pipe.ReadFD(), EventLoop::Event_Read, [this](int event) { onPipeEvent(); }) == -1) {
      throw std::runtime_error("Add pipe fd to poller failed");
    }
  }

  EventLoop::EventLoop(std::string name) {
    _name = std::move(name);
#if defined(HAS_EPOLL)
    _epoll_fd = epoll_create(EPOLL_SIZE);
  if (_epoll_fd == -1) {
      throw runtime_error(StrPrinter << "Create epoll fd failed: " << get_uv_errmsg());
  }
  SockOps::setCloExec(_epoll_fd);
#endif //HAS_EPOLL
    _logger = Logger::Instance().shared_from_this();
    addEventPipe();
  }

  void EventLoop::shutdown() {
    async_l([]() {
      throw ExitException();
    }, false, true);

    if (_loop_thread) {
      //防止作为子进程时崩溃
      try { _loop_thread->join(); } catch (...) {}
      delete _loop_thread;
      _loop_thread = nullptr;
    }
  }

  EventLoop::~EventLoop() {
    shutdown();
#if defined(HAS_EPOLL)
    if (_epoll_fd != -1) {
      close(_epoll_fd);
      _epoll_fd = -1;
  }
#endif //defined(HAS_EPOLL)
    //退出前清理管道中的数据
    onPipeEvent();
    LOG_INFO << this;
  }

  int EventLoop::AddEvent(int fd, int event, PollEventCallBack callback) {
    TimeTicker();
    if (!callback) {
      LOG_WARN << "PollEventCB is IsEmpty";
      return -1;
    }

    if (isCurrentThread()) {
#if defined(HAS_EPOLL)
      struct epoll_event ev = {0};
      ev.events = (toEpoll(event)) | EPOLLEXCLUSIVE;
      ev.data.fd = fd;
      int ret = epoll_ctl(_epoll_fd, EPOLL_CTL_ADD, fd, &ev);
      if (ret == 0) {
          _event_map.emplace(fd, std::make_shared<PollEventCB>(std::move(cb)));
      }
      return ret;
#else
#ifndef _WIN32
      //win32平台，socket套接字不等于文件描述符，所以可能不适用这个限制
      if (fd >= FD_SETSIZE || _event_map.size() >= FD_SETSIZE) {
        LOG_WARN << "select() can not watch fd bigger than " << FD_SETSIZE;
        return -1;
      }
#endif
      auto record = std::make_shared<PollRecord>();
      record->m_event = event;
      record->m_call_back = std::move(callback);
      _event_map.emplace(fd, record);
      return 0;
#endif //HAS_EPOLL
    }

    Async([this, fd, event, callback]() {
      AddEvent(fd, event, std::move(const_cast<PollEventCallBack &>(callback)));
    });
    return 0;
  }

  int EventLoop::RemoveEvent(int fd, PollDelCallBack callback) {
    TimeTicker();
    if (!callback) {
      callback = [](bool success) {};
    }

    if (isCurrentThread()) {
#if defined(HAS_EPOLL)
      bool success = epoll_ctl(_epoll_fd, EPOLL_CTL_DEL, fd, nullptr) == 0 && _event_map.erase(fd) > 0;
      cb(success);
      return success ? 0 : -1;
#else
      callback(_event_map.erase(fd));
      return 0;
#endif //HAS_EPOLL

    }

    //跨线程操作
    Async([this, fd, callback]() {
      RemoveEvent(fd, std::move(const_cast<PollDelCallBack &>(callback)));
    });
    return 0;
  }

  int EventLoop::modifyEvent(int fd, int event) {
    TimeTicker();
#if defined(HAS_EPOLL)
    struct epoll_event ev = {0};
  ev.events = toEpoll(event);
  ev.data.fd = fd;
  return epoll_ctl(_epoll_fd, EPOLL_CTL_MOD, fd, &ev);
#else
    if (isCurrentThread()) {
      auto it = _event_map.find(fd);
      if (it != _event_map.end()) {
        it->second->m_event = event;
      }
      return 0;
    }
    Async([this, fd, event]() {
      modifyEvent(fd, event);
    });
    return 0;
#endif //HAS_EPOLL
  }

  Task::Ptr EventLoop::Async(TaskIn task, bool may_sync) {
    return async_l(std::move(task), may_sync, false);
  }

  Task::Ptr EventLoop::AsyncFirst(TaskIn task, bool may_sync) {
    return async_l(std::move(task), may_sync, true);
  }

  Task::Ptr EventLoop::async_l(TaskIn task, bool may_sync, bool first) {
    TimeTicker();
    if (may_sync && isCurrentThread()) {
      task();
      return nullptr;
    }

    auto ret = std::make_shared<Task>(std::move(task));
    {
      std::lock_guard<std::mutex> lck(_mtx_task);
      if (first) {
        _list_task.emplace_front(ret);
      } else {
        _list_task.emplace_back(ret);
      }
    }
    //写数据到管道,唤醒主线程
    _pipe.Write("", 1);
    return ret;
  }

  bool EventLoop::isCurrentThread() {
    return !_loop_thread || _loop_thread->get_id() == std::this_thread::get_id();
  }

  inline void EventLoop::onPipeEvent() {
    char buf[1024];
    int err = 0;
    while (true) {
      if ((err = _pipe.Read(buf, sizeof(buf))) > 0) {
        // 读到管道数据,继续读,直到读空为止
        continue;
      }
      if (err == 0 || GetError(true) != RD_EAGAIN) {
        // 收到eof或非EAGAIN(无更多数据)错误,说明管道无效了,重新打开管道
        LOG_ERROR << "Invalid pipe fd of event poller, reopen it";
        RemoveEvent(_pipe.ReadFD());
        _pipe.ReOpen();
        addEventPipe();
      }
      break;
    }

    decltype(_list_task) _list_swap;
    {
      std::lock_guard <std::mutex> lck(_mtx_task);
      _list_swap.swap(_list_task);
    }

    _list_swap.For_Each([&](const Task::Ptr &task) {
      try {
        (*task)();
      } catch (ExitException &) {
        _exit_flag = true;
      } catch (std::exception &ex) {
        LOG_ERROR << "Exception occurred when do Async task: " << ex.what();
      }
    });
  }

  BufferRaw::Ptr EventLoop::getSharedBuffer() {
    auto ret = _shared_buffer.lock();
    if (!ret) {
      //预留一个字节存放\0结尾符
      ret = BufferRaw::Create();
      ret->SetCapacity(1 + SOCKET_DEFAULT_BUF_SIZE);
      _shared_buffer = ret;
    }
    return ret;
  }

  std::thread::id EventLoop::getThreadId() const {
    return _loop_thread ? _loop_thread->get_id() : std::thread::id();
  }

  const std::string &EventLoop::getThreadName() const {
    return _name;
  }

  static thread_local std::weak_ptr<EventLoop> s_current_poller;

// static
  EventLoop::Ptr EventLoop::getCurrentPoller() {
    return s_current_poller.lock();
  }

  void EventLoop::runLoop(bool blocked, bool ref_self) {
    if (blocked) {
      if (ref_self) {
        s_current_poller = shared_from_this();
      }
      _sem_run_started.Post();
      _exit_flag = false;
      uint64_t minDelay;
#if defined(HAS_EPOLL)
      struct epoll_event events[EPOLL_SIZE];
      while (!_exit_flag) {
          minDelay = getMinDelay();
          startSleep();//用于统计当前线程负载情况
          int ret = epoll_wait(_epoll_fd, events, EPOLL_SIZE, minDelay ? minDelay : -1);
          sleepWakeUp();//用于统计当前线程负载情况
          if (ret <= 0) {
              //超时或被打断
              continue;
          }
          for (int i = 0; i < ret; ++i) {
              struct epoll_event &ev = events[i];
              int fd = ev.data.fd;
              auto it = _event_map.find(fd);
              if (it == _event_map.end()) {
                  epoll_ctl(_epoll_fd, EPOLL_CTL_DEL, fd, nullptr);
                  continue;
              }
              auto cb = it->second;
              try {
                  (*cb)(toPoller(ev.events));
              } catch (std::exception &ex) {
                  LOG_ERROR << "Exception occurred when do event task: " << ex.what();
              }
          }
      }
#else
      int ret, max_fd;
      FdSet set_read, set_write, set_err;
      List<PollRecord::Ptr> callback_list;
      struct timeval tv;
      while (!_exit_flag) {
        //定时器事件中可能操作_event_map
        minDelay = getMinDelay();
        tv.tv_sec = (decltype(tv.tv_sec)) (minDelay / 1000);
        tv.tv_usec = 1000 * (minDelay % 1000);

        set_read.fdZero();
        set_write.fdZero();
        set_err.fdZero();
        max_fd = 0;
        for (auto &pr: _event_map) {
          if (pr.first > max_fd) {
            max_fd = pr.first;
          }
          if (pr.second->m_event & Event_Read) {
            set_read.fdSet(pr.first);//监听管道可读事件
          }
          if (pr.second->m_event & Event_Write) {
            set_write.fdSet(pr.first);//监听管道可写事件
          }
          if (pr.second->m_event & Event_Error) {
            set_err.fdSet(pr.first);//监听管道错误事件
          }
        }

        StartSleep();//用于统计当前线程负载情况
        ret = rd_select(max_fd + 1, &set_read, &set_write, &set_err, minDelay ? &tv : nullptr);
        SleepWakeUp();//用于统计当前线程负载情况

        if (ret <= 0) {
          //超时或被打断
          continue;
        }
        //收集select事件类型
        for (auto &pr: _event_map) {
          int event = 0;
          if (set_read.isSet(pr.first)) {
            event |= Event_Read;
          }
          if (set_write.isSet(pr.first)) {
            event |= Event_Write;
          }
          if (set_err.isSet(pr.first)) {
            event |= Event_Error;
          }
          if (event != 0) {
            pr.second->m_attach = event;
            callback_list.emplace_back(pr.second);
          }
        }

        callback_list.For_Each([](PollRecord::Ptr &record) {
          try {
            record->m_call_back(record->m_attach);
          } catch (std::exception &ex) {
            LOG_ERROR << "Exception occurred when do event task: " << ex.what();
          }
        });
        callback_list.clear();
      }
#endif //HAS_EPOLL
    } else {
      _loop_thread = new std::thread(&EventLoop::runLoop, this, true, ref_self);
      _sem_run_started.Wait();
    }
  }

  uint64_t EventLoop::flushDelayTask(uint64_t now_time) {
    decltype(_delay_task_map) task_copy;
    task_copy.swap(_delay_task_map);

    for (auto it = task_copy.begin(); it != task_copy.end() && it->first <= now_time; it = task_copy.erase(it)) {
      //已到期的任务
      try {
        auto next_delay = (*(it->second))();
        if (next_delay) {
          //可重复任务,更新时间截止线
          _delay_task_map.emplace(next_delay + now_time, std::move(it->second));
        }
      } catch (std::exception &ex) {
        LOG_ERROR << "Exception occurred when do delay task: " << ex.what();
      }
    }

    task_copy.insert(_delay_task_map.begin(), _delay_task_map.end());
    task_copy.swap(_delay_task_map);

    auto it = _delay_task_map.begin();
    if (it == _delay_task_map.end()) {
      //没有剩余的定时器了
      return 0;
    }
    //最近一个定时器的执行延时
    return it->first - now_time;
  }

  uint64_t EventLoop::getMinDelay() {
    auto it = _delay_task_map.begin();
    if (it == _delay_task_map.end()) {
      //没有剩余的定时器了
      return 0;
    }
    auto now = GetCurrentMillisecond();
    if (it->first > now) {
      //所有任务尚未到期
      return it->first - now;
    }
    //执行已到期的任务并刷新休眠延时
    return flushDelayTask(now);
  }

  EventLoop::DelayTask::Ptr EventLoop::doDelayTask(uint64_t delay_ms, std::function<uint64_t()> task) {
    DelayTask::Ptr ret = std::make_shared<DelayTask>(std::move(task));
    auto time_line = GetCurrentMillisecond() + delay_ms;
    AsyncFirst([time_line, ret, this]() {
      //异步执行的目的是刷新select或epoll的休眠时间
      _delay_task_map.emplace(time_line, ret);
    });
    return ret;
  }

///////////////////////////////////////////////




RD_NAMESPACE_END